package com.icbc.dccsh.bootstrap;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.XMLConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.icbc.dccsh.logpumper.Core;
import com.icbc.dccsh.logpumper.IConfigurable;
import com.icbc.dccsh.writter.KafkaWritter;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

public class RabbitMQBootstrap extends Bootstrap implements IConfigurable {
	public static final Logger LOG = LoggerFactory.getLogger(RabbitMQBootstrap.class);
	
	private String _host;
	private int _port;
	private String _queue;
	
	public void initialize(){
		// Nothing to do.
	}
	public void destructor(){
		// Nothing to do.
	}
	public void parseConfig(){
		XMLConfiguration config;
		try {
			config = new XMLConfiguration("conf.xml");
			this._host = config.getString("rabbitmq.host", "localhost1");
			this._port = config.getInt("rabbitmq.port", 5672);
			this._queue = config.getString("rabbitmq.queue", "logpumper_q");
		} catch (ConfigurationException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	public void boot() {
		String queueName = this._queue;
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost(this._host);
		factory.setPort(this._port);
		try {
			Connection connection = factory.newConnection();
			final Channel channel = connection.createChannel();
			channel.queueDeclare(queueName, false, false, false, null);
			LOG.info(" [*] Waiting for messages. To exit press CTRL+C");

			Consumer consumer = new DefaultConsumer(channel) {
				@Override
				public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
						byte[] body) throws IOException {
					String message = new String(body, "UTF-8");
					LOG.info(" [x] Received '" + message + "'");
					try {
						doWork(message);
					} finally {
						LOG.info(" [x] Done");
						channel.basicAck(envelope.getDeliveryTag(), false);
					}
				}
				
				private void doWork(String message){
					LOG.info("Start to do work...");
					Core c = new Core();
					c.start();
					LOG.info("End work...");
				}
			};
			channel.basicConsume(queueName, false, consumer);
//			channel.basicQos(1);
//			channel.basicConsume(queueName, true, consumer);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (TimeoutException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}
